package com.ybzl.system.config;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ybzl.common.utils.DateUtils;
import com.ybzl.system.domain.IotData;
import com.ybzl.system.domain.IotDevice;
import com.ybzl.system.service.IotDataService;
import com.ybzl.system.service.IotService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.*;

/***
 *
 * 实现断线重连包含两种方式（重连以后需要重新订阅一下主题，）
 *  1、基于框架实现的自动重连
 *  2、基于链接丢失时事件回调下的手动重连
 *
 *  todo: 需要验证一下自动重连和手动重连时客户端id的取值
 *  todo: 在由ping失败时导致的重连验证，以上两种思路是否都可行
 *
 *
 * * 从 Qos 上入手：
 *   * QoS 0 最多交付一次，消息可能丢失；
 *  * QoS 1 至少交付一次，消息可以保证到达，但是可能重复；
 *  * QoS 2 只交付一次，消息保证到达，并且不会重复。
 *
 */
@Slf4j
// @Component
public class EmqxConfig {

    private static final Logger logger = LoggerFactory.getLogger(EmqxConfig.class);
    @Value("${serverUri}")
    private String serverUri;

    @Autowired
    private RedisTemplate<Object, Object> redisTemplate;
    /**
     * 客户端Id
     */
    private String clientId = "iot_service@" + System.currentTimeMillis();

    /**
     * 客户端对象
     */
    private static MqttClient mqttClient;

    @Autowired
    private IotService iotService;

    @Autowired
    private IotDataService iotDataService;

    @PostConstruct
    public void init() {
        try {
            MemoryPersistence memoryPersistence = new MemoryPersistence();
            mqttClient = new MqttClient(serverUri, clientId, memoryPersistence);
            MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
            mqttConnectOptions.setKeepAliveInterval(20);

            // 自动重连
            mqttConnectOptions.setAutomaticReconnect(true);
            //mqttConnectOptions.setMaxReconnectDelay(128000);

            // cleanSession 为 false 表示断线后会话不会被清除，服务器会保留客户端订阅信息等待客户端重连，在断线期间发布的消息也会保留，当同一客户端重连后会自动订阅原订阅的主题，并且可以订阅到断线时发布的主题消息（ 同一clientId 表示同一个客户端 ）。
            mqttConnectOptions.setCleanSession(true);

            // 连接等待超时时长，0不限时
            mqttConnectOptions.setConnectionTimeout(0);
            mqttClient.connect(mqttConnectOptions);
            mySucribe();

            mqttClient.setCallback(new MqttCallbackExtended() {


                @Override
                public void connectionLost(Throwable throwable) {
                    throwable.printStackTrace();
                    // 连接断开
                    logger.info("[MQTT] 连接断开，10S之后尝试重连...");
                    /*while (true) {
                        try {
                            Thread.sleep(10000);
                            // 重新连接
                            mqttClient.reconnect();

                            // 重新订阅
                            //client.subscribe(topic, QOS);
                            break;
                        } catch (Exception e) {
                            e.printStackTrace();
                            continue;
                        }
                    }*/
                }

                @Override
                public void messageArrived(String topic, MqttMessage mqttMessage) throws IOException {
                    logger.error("收到消息：{}:{}", topic, mqttMessage.getPayload());
                }
                /**
                 * deliveryComplete() 方法实现于 MqttCallback 接口，在客户端发送消息至服务器成功时触发
                 * @param token the delivery token associated with the message.
                 */
                // 在deliveryComplete回调函数中，我们可以处理已发送消息的确认信息，例如记录日志、错误处理等操作。
                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    try {
                        // 消息发送成功的回调
                        int messageId = token.getMessageId();
                        // 可以根据messageId进行相关操作
                    } catch (Exception e) {
                        logger.error("{}", e.getMessage(), e);
                    }

                }

                // 1、断线重连下的回调机制
                @Override
                public void connectComplete(boolean reconnect, String serverUri) {
                    // 客户端连接成功
                    logger.info("[MQTT] 连接成功，重新订阅主题...{}:{}",reconnect,serverUri);
                    try {
                        // client.subscribe(topic, QOS);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }

            });
        } catch (Exception e) {
            log.error("MQTT服务异常.{}", e.getMessage(), e);
        }
    }

    private void mySucribe() throws MqttException {

        // 组装所有盒子的订阅topic
        List<String> list = new ArrayList<>();
        List<IotDevice> devices = iotService.devices();
        devices.forEach(tmp -> {
            String topic = "sys/device/" + tmp.getCode() + "/variant_data";
            list.add(topic);
        });


        String[] strArrStrings = new String[list.size()];
        int[] qosArr = new int[list.size()];
        for (int i = 0; i <= list.size() - 1; i++) {
            strArrStrings[i] = (String) list.get(i);
            qosArr[i] = 2;
        }
        mqttClient.subscribe(strArrStrings, qosArr);
    }

   /* public static void push(String topic, List<String> userList, Object message) {

        Map<String, Object> msgRoleMap = redisTemplate.opsForHash().entries("msgRole");
        for (String userId : userList) {
            String temp = null;
            String value = (String) msgRoleMap.get(userId);
            MqttMessage sendMessage = new MqttMessage();
            sendMessage.setQos(1);
            Map<String, Object> resultMap = new HashMap<>();

            temp = topic + userId;
            resultMap.put("topic", temp);
            resultMap.put("data", message);
            sendMessage.setPayload(JSONObject.toJSONString(resultMap).getBytes());
            try {
                mqttClient.publish(temp, sendMessage);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }

    }*/

    public static void push(String topic, boolean isReceive, Object message) {
        MqttMessage sendMessage = new MqttMessage();
        sendMessage.setQos(1);
        Map<String, Object> resultMap = new HashMap<>();
        try {
            if (isReceive) {
                sendMessage.setPayload(new ObjectMapper().writeValueAsBytes(message));
            } else {
                resultMap.put("topic", topic);
                resultMap.put("data", message);
                sendMessage.setPayload(new ObjectMapper().writeValueAsBytes(resultMap));
            }
            mqttClient.publish(topic, sendMessage);
        } catch (MqttException e) {
            log.error("{}", e.getMessage(), e);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }

    public static void push(String topic, Object message) {
        push(topic, false, message);
    }

    public static void pushSc(String topic, int num, boolean isReceive, Object message, boolean isf) {
        MqttMessage sendMessage = new MqttMessage();
        sendMessage.setQos(num);
        // 是否保留最后一条消息保留(true:保留；false:不保留)
        sendMessage.setRetained(isf);
        Map<String, Object> resultMap = new HashMap<>();
        try {
            if (isReceive) {
                sendMessage.setPayload(new ObjectMapper().writeValueAsBytes(message));
            } else {
                resultMap.put("topic", topic);
                resultMap.put("data", message);
                sendMessage.setPayload(new ObjectMapper().writeValueAsBytes(resultMap));
            }
            mqttClient.publish(topic, sendMessage);
        } catch (Exception e) {
            log.error("mqtt超时.{}", e.getMessage(), e);
        }
    }

    /**
     * 删除保留最后一条消息
     *
     * @param topic
     * @param Qos
     */
    public static void delet(String topic, int Qos) {
        MqttMessage sendMessage = new MqttMessage();
        sendMessage.setQos(Qos);
        // 是否保留最后一条消息保留
        sendMessage.setRetained(true);
        sendMessage.setPayload(new byte[0]);
        try {
            mqttClient.publish(topic, sendMessage);
        } catch (Exception e) {
            log.error("mqtt超时.{}", e.getMessage(), e);
        }
    }

}

